-module(rtps_submsg).

%%% This module is used to convert outgoing submessages from the
%%% readers and writers to the format used by the `rtps_sender'. (This
%%% is not yet the wrre format.)

-include("rtps.hrl").
-include("rtps_psm.hrl").
-include("rtps_history_cache.hrl").

-export([data/5, gap/3, heartbeat/7, acknack/6]).

-spec heartbeat(Writer_guid, Reader_guid, First_sn, Last_sn, Count,
		Final_flag, Liveliness_flag) -> rtps:heartbeat() when
      Writer_guid :: rtps:guid(),
      Reader_guid :: rtps:guid(),
      First_sn :: rtps:sequence_number(),
      Last_sn :: rtps:sequence_number(),
      Count :: rtps:count(),
      Final_flag :: boolean(),
      Liveliness_flag :: boolean().
%% @doc Using the parameters supplied, create a internal
%% representation of a heartbeat message that can be processed by the
%% `rtps_sender'.
heartbeat(#guid{entity_id = Writer_entity_id} = Writer_guid,
	  #guid{entity_id = Reader_entity_id} = Reader_guid,
	  First_sn, Last_sn, Count, Final_flag, Liveliness_flag) ->
    Submsg = #heartbeat{writer_id = Writer_entity_id, reader_id = Reader_entity_id,
			first_sn = First_sn, last_sn = Last_sn, count = Count,
			final_flag = Final_flag, liveliness_flag = Liveliness_flag},
    #msg{src_guid = Writer_guid, dst_guid = Reader_guid, submsg = Submsg}.
    
-spec data(Writer_guid, Reader_guid, Writer_cache, Data_sns, Inline_qos) -> Data_msgs when
      Writer_guid :: rtps:guid(),
      Reader_guid :: rtps:guid(),
      Writer_cache :: pid(),
      Data_sns :: [rtps:sequence_number()],
      Inline_qos :: undefined | [rtps_qos:qos()],
      Data_msgs :: [rtps:data_msg()].
%% @doc Given a list of cache changes represented by their sequence
%% numbers, create a list of internal representations of data
%% submessages ready to be proceesed by the `rtps_sender'. This
%% function fetches the changes from the writer's history cache.
%%
%% TODO: Optimize by eliminating repeatedly calling the history cache.
data(#guid{entity_id = Writer_entity_id} = Writer_guid,
     #guid{entity_id = Reader_entity_id} = Reader_guid,
     Writer_cache, Seq_nums, Inline_qos) ->
    Cache_changes = [rtps_history_cache:get_change(Writer_cache, Writer_guid, Seq_num) || Seq_num <- Seq_nums],
    Fun = fun(#cache_change{sequence_number = Seq_num, data_value = undefined,
			    instance_handle = Serialized_key, size = Size}, Acc) ->
		  Submsg = #data_msg{writer_id = Writer_entity_id, reader_id = Reader_entity_id,
				     writer_sn = Seq_num, inline_qos = Inline_qos,
				     serialized_data = undefined,
				     serialized_key = Serialized_key},
		  [#msg{src_guid = Writer_guid, dst_guid = Reader_guid, submsg = Submsg,
			size = Size} | Acc];
	     (#cache_change{sequence_number = Seq_num, data_value = Serialized_data,
			    instance_handle = _Serialized_key, size = Size,
			    timestamp = Timestamp}, Acc) ->
		  Submsg = #data_msg{writer_id = Writer_entity_id, reader_id = Reader_entity_id,
				     writer_sn = Seq_num, inline_qos = Inline_qos,
				     serialized_data = Serialized_data,
				     serialized_key = undefined},
		  [#msg{src_guid = Writer_guid, dst_guid = Reader_guid, submsg = Submsg,
			size = Size,  timestamp = Timestamp} | Acc];
	     (false, Acc) ->
		  Acc
	  end,
    Data_msgs = lists:foldl(Fun, [], Cache_changes),
    lists:reverse(Data_msgs).

-spec gap(Writer_guid, Reader_guid, Sequence_numbers) -> [Gap] when
      Writer_guid :: rtps:guid(),
      Reader_guid :: rtps:guid(),
      Sequence_numbers :: [rtps:sequence_number()],
      Gap :: rtps:gap().
%% A gap message is used to communicate that a series of sequence
%% numbers is not relevant. Internally, a gap message may contain a
%% long list of not relevant sequence numbers, but externally, this
%% list is limited to a maximum size, and therefor the result is a
%% list of gap messages. A gap message has the following structure:
%% 
%% The irrelevant sequence numbers communicated by the Gap message are
%% composed of two groups:
%% 1. All sequence numbers in the range gapStart <= sequence_number <=
%% gapList.base -1
%% 2. All the sequence numbers that appear explicitly listed in the
%% gapList.
%% 
%% So, we have to deal both with a long list of sequence numbers and
%% the case in which sequence numbers are consecutive.
gap(_Writer_guid, _Reader_guid, []) ->
    [];
gap(Writer_guid, Reader_guid, [Gap_start | Rest]) ->
    gap(Writer_guid, Reader_guid, Gap_start, Rest, Gap_start + 1, [], []).

%% TODO: Maybe one can optimize the way consecutive sequence numbers
%% are taken together. For example, Currently, a single sequence
%% number missing in the list will make the sequence numbers following
%% the missing one being added to the gapList set, but it might be
%% better to start a new GAP message.
gap(_Writer_guid, _Reader_guid, _Gap_start, [], _N, _Set, Acc) ->
    lists:reverse(Acc);
%% Next sequence number in the list is consecutive to the previous one
gap(Writer_guid, Reader_guid, Gap_start, [N | Rest], N, _Set, Acc) ->
    gap(Writer_guid, Reader_guid, Gap_start, Rest, N + 1, [], Acc);
%% Put the non consecutive sequence numbers in the set. Numbers in the
%% set have a value relative to the Base value: i.e. the first number
%% in the set is the base, the second number is Base + 1 etc. The size
%% of the set should not exceed the PSM defined maximum.
gap(Writer_guid, Reader_guid, Gap_start, [N | Rest], Base, Set, Acc)
  when N =< Base + ?MAX_NS_SIZE ->
    gap(Writer_guid, Reader_guid, Gap_start, Rest, Base, [N | Set], Acc);
%% We've collected a GAP! Turn collected data into a gap message and
%% continue with the rest of the list of irrelevant sequencer numbers.
gap(#guid{entity_id = Writer_entity_id} = Writer_guid,
    #guid{entity_id = Reader_entity_id} = Reader_guid, Gap_start, [N | Rest], Base, Set, Acc) ->
    {Gap_list, _Rest} = rtps_psm:to_sequence_number_set(Base, Set),
    Submsg = #gap{writer_id = Writer_entity_id, reader_id = Reader_entity_id,
		  gap_start = Gap_start, gap_list = Gap_list},
    Acc1 = [#msg{src_guid = Writer_guid, dst_guid = Reader_guid, submsg = Submsg} | Acc],
    gap(Writer_entity_id, Reader_entity_id, N, Rest, N + 1, [], Acc1).

-spec acknack(Reader_guid, Writer_guid, Avail_seq_num, Missing_seq_nums, Count, Final_flag) -> rtps:acknack() when
      Reader_guid :: rtps:guid(),
      Writer_guid :: rtps:guid(),
      Avail_seq_num :: rtps:sequence_number(),
      Missing_seq_nums :: [rtps:sequence_number()],
      Count :: rtps:count(),
      Final_flag :: boolean().
%% TODO: If the sequence numbers in the list of missing numbers are
%% not within the allowed range, they are currently silently dropped,
%% which should not be a problem because the reader will re-request
%% these missing changes again later after the preceding missing
%% changes have been fixed. How to deal more effectively with this
%% situation?
acknack(#guid{entity_id = Reader_entity_id} = Reader_guid,
	#guid{entity_id = Writer_entity_id} = Writer_guid,
	Avail_seq_num, Missing_seq_nums, Count, Final_flag) ->
    Base = Avail_seq_num + 1,
    {Reader_sn_state, _Rest} = rtps_psm:to_sequence_number_set(Base, Missing_seq_nums),
    Submsg = #acknack{writer_id = Writer_entity_id, reader_id = Reader_entity_id,
		      reader_sn_state = Reader_sn_state, final_flag = Final_flag, count = Count},
    #msg{src_guid = Reader_guid, dst_guid = Writer_guid, submsg = Submsg}.


%%====================================================================
%% Internal functions
%%====================================================================
